package promote.brozen.netty.jav.nio;

import org.apache.commons.collections4.CollectionUtils;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.*;
import java.util.function.BiConsumer;

/**
 * @author Brozen
 * @date 2020/9/27 8:21 AM
 * @email brozen@qq.com
 */
public class SelectorReactor implements Runnable {

    protected Selector selector;

    protected BiConsumer<SelectionKey, SelectableChannel> selectCallback;

    public SelectorReactor() throws IOException {
        this(Selector.open());
    }

    public SelectorReactor(Selector selector) {
        this.selector = selector;
    }

    /**
     * 添加通道到Reactor中，并注册Selector
     */
    public void register(SelectableChannel channel, int opt) {
        try {
            channel.register(selector, opt, this);
        } catch (ClosedChannelException e) {
            throw new IllegalStateException("Channel closed", e);
        }
    }

    @Override
    public void run() {
        try {
            while (true) {
                // 每1s查看一次是否有可读channel
                if (selector.select(1000) <= 0) {
                    continue;
                }

                // 获取到可读通道
                Set<SelectionKey> keys = selector.selectedKeys();
                if (CollectionUtils.isEmpty(keys)) {
                    continue;
                }

                // 读出channel，交给callback处理
                Iterator<SelectionKey> iterator = keys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if (!key.isValid()) {
                        continue;
                    }

                    // 已就绪，交给回调处理
                    SelectableChannel channel = key.channel();
                    this.selectCallback.accept(key, channel);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new IllegalStateException(e);
        }
    }

}
